import java.sql.{Connection, DriverManager, PreparedStatement}

object SparkMlib {
  private val jdbcUrl = "jdbc:mysql://192.168.244.11:3306/stats?serverTimezone=Asia/Shanghai"
  private val username = "root"
  private val password = "123456"

  def main(args: Array[String]): Unit = {
   /// System.setProperty("hadoop.home.dir", "C:\\hadoop")

    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("StudentGPACollaborativeFiltering")
      .set("spark.sql.warehouse.dir", "file:///C:/spark-warehouse")
      .set("spark.driver.extraClassPath", "D:/develop/apache-maven-3.6.1/repository/mysql/mysql-connector-java/8.0.22/mysql-connector-java-8.0.22.jar")
      .set("spark.driver.memory", "4g")
      .set("spark.executor.memory", "4g")

    val sc = new SparkContext(conf)
    sc.setLogLevel("INFO")
    val sqlContext = new SQLContext(sc)

    try {
      // 1. 数据加载与预处理
      println("开始从MySQL加载学生绩点数据...")
      val startTime = System.currentTimeMillis()

      val inputDF = sqlContext.read
        .format("jdbc")
        .option("url", jdbcUrl)
        .option("dbtable", "student_info")
        .option("user", username)
        .option("password", password)
        .load()
        .select("student_id", "name", "semester", "gpa")

      println(s"数据加载完成，耗时: ${(System.currentTimeMillis() - startTime)/1000.0}秒")
      println("输入数据Schema:")
      inputDF.printSchema()
      println(s"总记录数: ${inputDF.count()}")
      inputDF.show(5)

      // 创建学生ID到姓名的映射
      val studentIdToName = inputDF.rdd.map { row =>
        val studentId = row.getAs[String]("student_id")
        val name = row.getAs[String]("name")
        (studentId, name)
      }.distinct().collectAsMap()

      // 广播学生ID到姓名的映射
      val studentIdToNameBC = sc.broadcast(studentIdToName)

      // 2. 数据清洗与转换
      println("数据清洗与转换...")
      val studentSemesterGPARDD = inputDF.rdd.map { row =>
        val studentId = row.getAs[String]("student_id")
        val semester = row.getAs[Int]("semester")
        val rawGpa = row.getAs[java.math.BigDecimal]("gpa").doubleValue()
        val gpa = math.min(4.0, math.max(0.0, rawGpa))
        (studentId, semester, gpa)
      }.filter { case (_, semester, gpa) =>
        semester >= 1 && semester <= 8 && gpa > 0
      }.cache()

      println(s"有效记录数: ${studentSemesterGPARDD.count()}")

      // 检查数据分布
      println("学生GPA统计信息:")
      studentSemesterGPARDD.map(_._3).stats()

      // ========== a. 建立用户-绩点关系矩阵 ==========
      println("构建用户-绩点关系矩阵...")
      val userGpaMatrix = studentSemesterGPARDD.map {
          case (studentId, semester, gpa) => (studentId, (semester, gpa))
        }.groupByKey()
        .mapValues(_.toMap)
        .cache()

      // 检查每个学生的学期数量
      println("每个学生的学期数量分布:")
      userGpaMatrix.map(_._2.size).stats()

      // 收集所有学期列表
      val allSemesters = (1 to 8).toSet

      // ========== b. 建立绩点-绩点关系矩阵（用户相似度矩阵） ==========
      println("计算用户相似度矩阵...")
      val similarityStartTime = System.currentTimeMillis()

      // 首先计算每个用户的平均GPA
      val userAvgGpa = userGpaMatrix.mapValues { semesterGpas =>
        semesterGpas.values.sum / semesterGpas.size
      }.collectAsMap()

      // 广播用户平均GPA
      val userAvgGpaBC = sc.broadcast(userAvgGpa)

      // 检查共同学期分布
      println("共同学期数量分布:")
      val commonSemesterStats = userGpaMatrix.cartesian(userGpaMatrix)
        .filter { case ((u1, _), (u2, _)) => u1 < u2 }
        .map { case ((_, s1), (_, s2)) => s1.keySet.intersect(s2.keySet).size }
        .stats()
      println(commonSemesterStats)

      // 优化后的用户相似度计算（已解决除零问题）
      val userSimilarities = {
        val userPairs = userGpaMatrix.cartesian(userGpaMatrix)
          .filter { case ((user1, _), (user2, _)) => user1 < user2 }

        userPairs.flatMap { case ((user1, semGpa1), (user2, semGpa2)) =>
            try {
              val commonSemesters = semGpa1.keySet.intersect(semGpa2.keySet)

              if (commonSemesters.nonEmpty) {
                val avg1 = userAvgGpaBC.value.getOrElse(user1, 0.0)
                val avg2 = userAvgGpaBC.value.getOrElse(user2, 0.0)

                var (covariance, stdDev1, stdDev2) = (0.0, 0.0, 0.0)

                commonSemesters.foreach { sem =>
                  val gpa1 = semGpa1.getOrElse(sem, 0.0)
                  val gpa2 = semGpa2.getOrElse(sem, 0.0)
                  val diff1 = gpa1 - avg1
                  val diff2 = gpa2 - avg2
                  covariance += diff1 * diff2
                  stdDev1 += diff1 * diff1
                  stdDev2 += diff2 * diff2
                }

                stdDev1 = sqrt(stdDev1)
                stdDev2 = sqrt(stdDev2)

                val similarity = if (stdDev1 * stdDev2 != 0) {
                  val sim = covariance / (stdDev1 * stdDev2)
                  // 限制相似度在[-1, 1]范围内
                  math.max(-1.0, math.min(1.0, sim))
                } else {
                  1.0 // 完全相同的用户
                }

                if (similarity > 0) {
                  Seq((user1, (user2, similarity)), (user2, (user1, similarity)))
                } else {
                  Seq.empty
                }
              } else {
                Seq.empty
              }
            } catch {
              case e: Exception =>
                println(s"计算用户 $user1 和 $user2 的相似度时出错: ${e.getMessage}")
                Seq.empty
            }
          }
          .groupByKey()
          .mapValues(_.toList.sortBy(-_._2).take(50)) // 每个用户保留最相似的50个用户
          .collectAsMap()
      }

      // 打印相似度统计信息
      println("相似度统计:")
      val similarityStats = userSimilarities.map { case (user, simUsers) =>
        if (simUsers.nonEmpty) {
          (simUsers.size, simUsers.head._2, simUsers.map(_._2).sum / simUsers.size)
        } else {
          (0, 0.0, 0.0)
        }
      }.toSeq

      println(s"平均每个用户有 ${similarityStats.map(_._1).sum / similarityStats.size} 个相似用户")
      println(s"最高相似度: ${similarityStats.map(_._2).max}")
      println(s"平均相似度: ${similarityStats.map(_._3).sum / similarityStats.size}")

      println(s"用户相似度计算完成，耗时: ${(System.currentTimeMillis() - similarityStartTime)/1000.0}秒")

      // 广播用户相似度矩阵
      val userSimilaritiesBC = sc.broadcast(userSimilarities)

      // ========== c. 基于协同过滤生成推荐列表 ==========
      println("生成推荐列表...")
      val recommendationStartTime = System.currentTimeMillis()

      // 首先收集所有用户的GPA数据
      val allUserGpas = userGpaMatrix.collectAsMap()

      val recommendations = userGpaMatrix.flatMap { case (userId, semesterGpas) =>
        // 获取当前用户未评分的学期
        val unratedSemesters = allSemesters.diff(semesterGpas.keySet)

        // 获取相似用户
        val similarUsers = userSimilaritiesBC.value.getOrElse(userId, Nil)

        unratedSemesters.map { targetSemester =>
          // 计算加权预测GPA
          var (weightedSum, weightSum) = (0.0, 0.0)
          var count = 0

          similarUsers.foreach { case (simUser, similarity) =>
            allUserGpas.get(simUser).foreach { simUserGpas =>
              simUserGpas.get(targetSemester).foreach { gpa =>
                weightedSum += gpa * similarity
                weightSum += similarity
                count += 1
              }
            }
          }

          // 计算预测GPA
          val predictedGPA = if (weightSum > 0) {
            math.min(4.0, math.max(0.0, weightedSum / weightSum))
          } else {
            // 如果没有相似用户的数据，使用该用户自己的平均GPA
            semesterGpas.values.sum / semesterGpas.size
          }

          (userId, targetSemester, predictedGPA)
        }
      }.cache()

      println(s"推荐生成完成，耗时: ${(System.currentTimeMillis() - recommendationStartTime)/1000.0}秒")
      println(s"生成的推荐总数: ${recommendations.count()}")

      // 打印推荐结果的统计信息
      println("推荐结果统计:")
      recommendations.map(_._3).stats()
      recommendations.take(10).foreach(println)

      // 6. 保存结果 - 修改为插入学生姓名
      println("保存推荐结果到数据库...")
      val saveStartTime = System.currentTimeMillis()

      recommendations.map { case (studentId, semester, gpa) =>
        val name = studentIdToNameBC.value.getOrElse(studentId, "未知学生")
        s"$name\t$semester\t${"%.2f".format(gpa)}"
      }.foreachPartition { partition =>
        var conn: Connection = null
        var stmt: PreparedStatement = null
        try {
          conn = DriverManager.getConnection(jdbcUrl, username, password)
          conn.setAutoCommit(false)

          // 修改SQL语句，使用name作为第一个字段
          stmt = conn.prepareStatement(
            "INSERT INTO gpa_recommendations1 (name, semester, predicted_gpa) VALUES (?, ?, ?) " +
              "ON DUPLICATE KEY UPDATE predicted_gpa = VALUES(predicted_gpa)"
          )

          var batchCount = 0
          partition.foreach { record =>
            val fields = record.split("\t")
            stmt.setString(1, fields(0))  // name
            stmt.setInt(2, fields(1).toInt)  // semester
            stmt.setDouble(3, fields(2).toDouble)  // predicted_gpa
            stmt.addBatch()
            batchCount += 1

            // 每100条执行一次批量提交
            if (batchCount % 100 == 0) {
              stmt.executeBatch()
              conn.commit()
            }
          }

          // 提交剩余批次
          stmt.executeBatch()
          conn.commit()
          println(s"成功插入/更新 $batchCount 条推荐记录")

        } catch {
          case e: Exception =>
            println(s"数据库操作错误: ${e.getMessage}")
            if (conn != null) conn.rollback()
        } finally {
          if (stmt != null) stmt.close()
          if (conn != null) conn.close()
        }
      }

      println(s"结果保存完成，总耗时: ${(System.currentTimeMillis() - saveStartTime)/1000.0}秒")
      println("处理完成！")

    } catch {
      case e: Exception =>
        println(s"程序执行出错: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      sc.stop()
    }
  }
}